package com.atuguigu.flink.Day01.Example;

import com.atuguigu.flink.Day01.Singlesensor.SensorSource;
import com.atuguigu.flink.sensor.SendsorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class MapExample {
    //只取出ID
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<SendsorReading> stream = env.addSource(new SensorSource());

        //匿名函数的返回值就说是输出的数据
        stream.map(r -> r.id).print();

        //匿名类写法
        stream.map(
                new MapFunction<SendsorReading, String>() {


                    @Override
                    public String map(SendsorReading sendsorReading) throws Exception {
                        return sendsorReading.id;
                    }
                }
        ).print();

        //刚好的方法
        stream.map(new IdExtractor()).print();




        stream.print();

        env.execute();
    }

    public  static  class IdExtractor implements MapFunction<SendsorReading,String>{

        @Override
        public String map(SendsorReading sendsorReading) throws Exception {
            return sendsorReading.id;
        }
    }
}
